package com.shujia.spark.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {


    val conf: SparkConf = new SparkConf()
      .setAppName("streaming")
      .setMaster("local[2]")


    /**
      * 创建streaming 上下文对象，指定batch的间隔时间,多久计算一次
      *
      */
    val ssc = new StreamingContext(conf, Durations.seconds(5))


    //设置checkpoint
    ssc.checkpoint("data/checkpoint")


    /**
      * ReceiverInputDStream: 被动接收数据，将接收过来的数据放在内存或者磁盘上
      * 接收数据会一直占用资源
      *
      * nc -lk 9888
      *
      *
      * yum install nc
      */

    //读取数据
    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)


    val wordsDS: DStream[String] = linesDS.flatMap(_.split(","))

    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1))

    /**
      * reduceByKey 只统计当前batch的数据
      *
      */
    //    val countDS: DStream[(String, Int)] = kvDS.reduceByKey(_ + _)


    /**
      *
      * @param seq    ：当前batch 每一个key所有的value
      * @param option : 之前batch累加计算你的结果
      * @return ： 返回最新的单词的数量
      */
    def updateFun(seq: Seq[Int], option: Option[Int]): Option[Int] = {

      //计算当前batch单词的数量
      val currCount: Int = seq.sum

      //获取之前单词的数量
      val lastCount: Int = option.getOrElse(0)

      //返回最新单词的数量
      Some(currCount + lastCount)

    }

    /**
      *
      * 有状态算子
      *
      * updateStateByKey： 每一个祭祀啊u你更新每一个key的状态（单词的数量）
      *
      *
      * 需要设置checkpoint路径，用于保存计算过程中的状态
      */


    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updateFun)


    //打印数据
    countDS.print()


    //启动spark streaming
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
